package com.atguigu.ting.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.ting.realtime.bean.TableProcess;
import com.atguigu.ting.realtime.util.DimUtil;
import com.atguigu.ting.realtime.util.JdbcUtil;
import com.atguigu.ting.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author lzc
 * @Date 2023/3/11 09:03
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    
    private Connection conn;
    private Jedis redisClient;
    
    // 每来一条数据, 这个方法就掉用一次
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t, Context context) throws Exception {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        
        String opType = data.getString("op_type");
        if ("delete".equals(opType)) {
            // 删除维度数据
            deleteData(data, tp);
        } else {
            // insert 和 update
            upsertData(data, tp);
        }
        
        // 操作缓存: 当是 delete 或者 update 的时候: 删除缓存
        if ("delete".equals(opType) || "update".equals(opType)) {
            // 删除缓存
            delCache(data, tp);
        }
        
    }
    
    private void delCache(JSONObject data, TableProcess tp) {
        // 获取 key
        String key = DimUtil.getRedisKey(tp.getSinkTable(), data.getString("id"));
        Long n = redisClient.del(key);
        if (n == 1) {
            System.out.println(key + " 删除成功");
        } else {
            System.out.println(key + " 删除失败");
        }
        
    }
    
    private void deleteData(JSONObject data, TableProcess tp) throws SQLException {
        // delete from user where id=?
        String sql = "delete from " + tp.getSinkTable() + " where id=?";
        PreparedStatement ps = conn.prepareStatement(sql);
        ps.setString(1, data.getString("id"));
        ps.execute();
        conn.commit();
        ps.close();
    }
    
    private void upsertData(JSONObject data, TableProcess tp) throws SQLException {
        // 1. 定义 sql 语句  upsert
        // upsert into user(a,b,c)values(?,?,?)
        StringBuilder sql = new StringBuilder();
        sql
            .append("upsert into ")
            .append(tp.getSinkTable())
            .append("(")
            .append(tp.getSinkColumns())
            .append(")values(")
            .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
            .append(")");
        System.out.println("phoenix 插入语句: " + sql);
        
        // 2. 获取预处理语句
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        // 3. 给sql 中的占位符进行赋值
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            System.out.println(column + "  " + data.getString(column));
            ps.setString(i + 1, data.getString(column));
        }
        // 4. 执行 sql 语句
        ps.execute();
        // 5. 手动提交
        conn.commit();
        // 6. 关闭预处理
        ps.close();
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 1. 创建 jdbc 连接
        conn = JdbcUtil.getPhoenixConnection();
        
        redisClient = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void close() throws Exception {
        JdbcUtil.closeConnection(conn);
        
    }
}
